package com.chukun.multi.masterslave;

import com.chukun.multi.utils.Debug;

import java.io.*;
import java.util.*;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author chukun
 * 主线程分发任务
 */
public class Master {

    private final String logFileBaseDir;
    private final String excludedOperationNames;
    private final String includedOperationNames;
    private final String destinationSysName;

    private static final int NUMBER_OF_FILES_FOR_EACH_DISPATCH = 5;

    private static final int WORKER_COUNT = Runtime.getRuntime().availableProcessors();

    /**
     * 利用builder模式
     * 解除多参数传递参数错误的问题，
     */
    public static class Builder {
        private String logFileBaseDir;
        private String excludedOperationNames;
        private String includedOperationNames;
        private String destinationSysName;


        public Builder setLogFileBaseDir(String logFileBaseDir) {
            this.logFileBaseDir = logFileBaseDir;
            return this;
        }

        public Builder setExcludedOperationNames(String excludedOperationNames) {
            this.excludedOperationNames = excludedOperationNames;
            return this;
        }


        public Builder setIncludedOperationNames(String includedOperationNames) {
            this.includedOperationNames = includedOperationNames;
            return this;
        }

        public Builder setDestinationSysName(String destinationSysName) {
            this.destinationSysName = destinationSysName;
            return this;
        }

        public Master builder() {
            return new Master(this);
        }
    }

    public Master(Builder builder) {
        this.logFileBaseDir = builder.logFileBaseDir;
        this.excludedOperationNames = builder.excludedOperationNames;
        this.includedOperationNames = builder.includedOperationNames;
        this.destinationSysName = builder.destinationSysName;
    }

    /**
     * 计算任务
     *
     * @param fileNamesReader
     * @return
     * @throws IOException
     */
    public Map<String, AtomicInteger> calculate(BufferedReader fileNamesReader) throws IOException {
        Map<String, AtomicInteger> repository = new ConcurrentSkipListMap<>();
        // 创建工作线程
        Worker[] workers = createAndStartWorkers(repository);
        // 指派任务给工作者线程
        dispatchTask(fileNamesReader, workers);

        // 等待工作者线程处理结束
        for (Worker worker : workers) {
            worker.terminate(true);
        }

        // 返回结果
        return repository;
    }

    /**
     * 创建工作线程
     *
     * @param repository
     * @return
     */
    private Worker[] createAndStartWorkers(Map<String, AtomicInteger> repository) {
        Worker[] workers = new Worker[WORKER_COUNT];
        Worker worker;
        Thread.UncaughtExceptionHandler eh = new Thread.UncaughtExceptionHandler() {
            @Override
            public void uncaughtException(Thread t, Throwable e) {
                e.printStackTrace();
            }
        };

        for (int i = 0; i < WORKER_COUNT; i++) {
            worker = new Worker(repository, excludedOperationNames, includedOperationNames, destinationSysName);
            workers[i] = worker;
            worker.setUncaughtExceptionHandler(eh);
            worker.start();
        }
        return workers;
    }

    /**
     * 分发任务
     *
     * @param fileNamesReader
     * @param workers
     * @throws IOException
     */
    private void dispatchTask(BufferedReader fileNamesReader, Worker[] workers) throws IOException {
        String line;
        Set<String> fileNames = new HashSet<>();
        int fileCount = 0;
        int workerIndex = -1;
        BufferedReader logFileReader;
        while ((line = fileNamesReader.readLine()) != null) {
            fileNames.add(line);
            fileCount++;
            if (0 == (fileCount % NUMBER_OF_FILES_FOR_EACH_DISPATCH)) {
                // 工作者线程间的负载均衡：采用简单的轮询选择worker
                workerIndex = (workerIndex + 1) % WORKER_COUNT;
                logFileReader = makeReaderFrom(fileNames);
                Debug.info("dispatch " + NUMBER_OF_FILES_FOR_EACH_DISPATCH + " files to worker :" + workerIndex);
                workers[workerIndex].submitWorkload(logFileReader);
                fileNames = new HashSet<>();
                fileCount = 0;
            }
        }
        if (fileCount > 0) {
            logFileReader = makeReaderFrom(fileNames);
            workerIndex = (workerIndex + 1) % WORKER_COUNT;
            workers[workerIndex].submitWorkload(logFileReader);
        }
    }

    /**
     * 拿到输入流
     *
     * @param logFileNames
     * @return
     */
    private BufferedReader makeReaderFrom(final Set<String> logFileNames) {
        BufferedReader logFileReader;

        InputStream in = new SequenceInputStream(new Enumeration<InputStream>() {

            private Iterator<String> it = logFileNames.iterator();

            @Override
            public boolean hasMoreElements() {
                return it.hasNext();
            }

            @Override
            public InputStream nextElement() {
                String fileName = it.next();
                InputStream in = null;
                try {
                    in = new FileInputStream(logFileBaseDir + fileName);
                } catch (FileNotFoundException e) {
                    throw new RuntimeException(e);
                }
                return in;
            }
        });
        logFileReader = new BufferedReader(new InputStreamReader(in));
        return logFileReader;
    }
}
